package event

import (
	"errors"
	"reflect"
	"sync"
)

var errBadChannel = errors.New("event: Subscribe argument does not have sendable channel type")

type Feed struct {
	once sync.Once  //ensures that only runs once
	sendLock  chan struct{}
	removeSub chan interface{}   //interrupts Send
	sendCases caseList //the active set of select cases used by send

	//the inbox holds  newly subscribes channels until they are added to SendCases
	mu sync.Mutex
	inbox caseList
	etype reflect.Type
}


const firstSubSendCase  = 1

type feedTypeError struct {
	got , want reflect.Type
	op string
}

func (e feedTypeError) Error() string {
	return "event: wrong type in " + e.op +" got "+e.got.String() + ", want " + e.want.String()
}


func (f * Feed) init(){
	f.removeSub = make(chan interface{})
	f.sendLock = make(chan struct{})
	f.sendLock <- struct{}{}
	f.sendCases = caseList{{Chan:reflect.ValueOf(f.removeSub),Dir:reflect.SelectRecv}}
}

func ( f * Feed)Subscribe(channel interface{}) Subscription{
	f.once.Do(f.init)

	chanval := reflect.ValueOf(channel)
	chantyp := chanval.Type()

	//判断接受的channel 是不是reflect.Chan类型 ，同时是不是发送管道类型
	if chantyp.Kind() != reflect.Chan || chantyp.ChanDir() & reflect.SendDir == 0{
		panic(errBadChannel)
	}

	//初始化feedsub
	sub := &feedSub{feed:f,channel:chanval,err:make(chan error,1)}

	f.mu.Lock()
	defer f.mu.Unlock()
	if !f.typecheck(chantyp.Elem()) {
		panic(feedTypeError{op: "Subscribe", got: chantyp, want: reflect.ChanOf(reflect.SendDir, f.etype)})
	}
	//Add the select case to the inbox
	// the next send will add it to f.sendCases
	cas := reflect.SelectCase{Dir:reflect.SelectSend , Chan:chanval}
	f.inbox = append(f.inbox, cas)
	return sub
}

func(f *Feed) typecheck(typ reflect.Type) bool {
	if f.etype == nil{
		f.etype = typ
		return true
	}

	return f.etype == typ
}


func (f * Feed)remove(sub * feedSub){
	ch := sub.channel.Interface()

	f.mu.Lock()
	index := f.inbox.find(ch)
	if index != - 1{
		f.inbox.delete(index)
		f.mu.Unlock()
	}

	f.mu.Lock()

	select {
	case f.removeSub <- ch:
		//把要移除订阅的管道传入
	case <- f.sendLock:
		f.sendCases = f.sendCases.delete(f.sendCases.find(ch))
		f.sendLock <- struct{}{}
	}
}

// Send delivers to all subscribed channels simultaneously.
// It returns the number of subscribers that the value was sent to.
func (f *Feed) Send(value interface{}) (nsent int) {
	rvalue := reflect.ValueOf(value)

	f.once.Do(f.init)
	<-f.sendLock

	// Add new cases from the inbox after taking the send lock.
	f.mu.Lock()
	f.sendCases = append(f.sendCases, f.inbox...)
	f.inbox = nil

	if !f.typecheck(rvalue.Type()) {
		f.sendLock <- struct{}{}
		f.mu.Unlock()
		panic(feedTypeError{op: "Send", got: rvalue.Type(), want: f.etype})
	}
	f.mu.Unlock()

	// Set the sent value on all channels.
	for i := firstSubSendCase; i < len(f.sendCases); i++ {
		//往所有注册的管道中发送数据
		f.sendCases[i].Send = rvalue
	}

	// Send until all channels except removeSub have been chosen.  给所有管道发信号，直到他们被移除
	// 'cases' tracks a prefix
	// of sendCases. When a send succeeds, the corresponding case moves to the end of
	// 'cases' and it shrinks by one element.
	cases := f.sendCases
	for {
		// Fast path: try sending without blocking before adding to the select set.
		// This should usually succeed if subscribers are fast enough and have free
		// buffer space.
		/* “cases”跟踪sendCases的前缀。当发送成功时，相应的case移到cases的末尾，并收缩一个元素。 */
		/*快速路径：在添加到选择集之前尝试不阻塞地发送。如果订阅者足够快并且有空闲的缓冲区空间，这通常会成功。*/
		for i := firstSubSendCase; i < len(cases); i++ {
			if cases[i].Chan.TrySend(rvalue) {
				nsent++
				cases = cases.deactivate(i)
				i--
			}
		}
		if len(cases) == firstSubSendCase {
			break
		}
		// Select on all the receivers, waiting for them to unblock.
		/* 监听所有管道 ，等待他们unblock*/
		/* 会尝试执行能执行的selectCase  执行成功返回choesn*/
		chosen, recv, _ := reflect.Select(cases)

		if chosen == 0 /* cases[0]里放着 <-f.removeSub  */ {
			index := f.sendCases.find(recv.Interface())
			f.sendCases = f.sendCases.delete(index)
			if index >= 0 && index < len(cases) {
				// Shrink 'cases' too because the removed case was still active.
				cases = f.sendCases[:len(cases)-1]
			}
		} else {
			cases = cases.deactivate(chosen)
			nsent++
		}
	}

	// Forget about the sent value and hand off the send lock.
	for i := firstSubSendCase; i < len(f.sendCases); i++ {
		f.sendCases[i].Send = reflect.Value{}
	}
	f.sendLock <- struct{}{}
	return nsent
}
//feedSubscribe
type feedSub struct {
	feed *Feed
	channel reflect.Value
	errOnce  sync.Once
	err chan error
}

func (sub *feedSub) Err() <-chan error {
	return sub.err
}


func (sub *feedSub)Unsubscribe(){
	sub.errOnce.Do(func(){

	})
}
type caseList []reflect.SelectCase

func (cs caseList) find(channel interface{}) int {
	for i,cas := range cs {
		if cas.Chan.Interface() == channel {
			return i
		}
	}

	return -1
}

func (cs caseList)delete (index int) caseList {
	return append(cs[:index],cs[index+1:]...)
}

func (cs caseList)deactivate(index int)caseList{
	last := len(cs) - 1
	cs[index] , cs[last] = cs[last] , cs[index]
	return cs[:last]
}
